
package backend

import (
	"fmt"
	"hash/crc32"
	"io"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	humanize "github.com/dustin/go-humanize"
	bolt "go.etcd.io/bbolt"
	"go.uber.org/zap"
)

var (
	defaultBatchLimit    = 10000
	defaultBatchInterval = 100 * time.Millisecond

	defragLimit = 10000

	// initialMmapSize是mmapped区域的初始大小。将其设置为大于
	// 可能的最大数据库大小可以防止写入程序阻塞读卡器。
	// 这只适用于linux。
	initialMmapSize = uint64(10 * 1024 * 1024 * 1024)

	// minSnapshotWarningTimeout是触发长时间运行的快照警告的最小阈值。
	minSnapshotWarningTimeout = 30 * time.Second
)

// Backend接口的主要功能就是将底层存储与上层进行解楠，其中定义底层存储需要对上层提供的接口
type Backend interface {
	// 创建一个只读事务，这里的ReadTx接口是v3存储对只读事务的抽象
	ReadTx() ReadTx
	// 创建一个批量事务，这里的BatchTx接口是对批量读写事务的抽象，
	BatchTx() BatchTx
	// ConcurrentReadTx返回非阻塞读取事务。
	ConcurrentReadTx() ReadTx
	// 创建快照，backend.snapshot实现了该Snapshot接口
	Snapshot() Snapshot
	Hash(ignores func(bucketName, keyName []byte) bool) (uint32, error)
	// 获取当前已存储的总字节数
	Size() int64
	// SizeInUse返回逻辑上正在使用的后端的当前大小。
	// 由于后端可以以非字节单位（如
	// 页数）管理可用空间，因此返回值的字节数可能不准确。
	SizeInUse() int64
	// OpenReadTxN返回后端当前打开的读取事务数。
	OpenReadTxN() int64
	Defrag() error	// 碎片整理
	ForceCommit()	// 提交批量读写事务
	Close() error
}

type Snapshot interface {
	// Size获取快照的大小。
	Size() int64
	// writet将快照写入给定的写入程序。
	WriteTo(w io.Writer) (n int64, err error)
	// 关闭关闭快照。
	Close() error
}

type txReadBufferCache struct {
	mu         sync.Mutex
	buf        *txReadBuffer
	bufVersion uint64
}

// 结构体backend是v3版本存储提供的Backend接口的默认实现，其底层存储使用的就是BoltDB
type backend struct {
	size int64	// 当前backend实例己存储的总字节数。
	// sizeInUse是在后端实际使用的字节数
	sizeInUse int64	//
	commits int64	// 从启动到目前为止，己经提交的事务数。
	// openReadTxN是当前在后端
	openReadTxN int64
	// mlock阻止交换后端数据库文件
	mlock bool

	mu sync.RWMutex
	db *bolt.DB	// 底层的BoltDB存储

	batchInterval time.Duration	// 两次批量读写事务提交的最大时间差。
	batchLimit    int	// 指定一次批量事务中最大的操作数，当超过该阔值时，当前的批量事务会自动提交
	// 批量读写事务，batchTxBuffered是在batchTx的基础上添加了缓存功能，两者都实现了BatchTx接口，
	batchTx       *batchTxBuffered

	readTx *readTx	// 只读事务，readTx实现了前面提到的ReadTx接口。
	// txReadBufferCache在“readTx”-readTx中镜像“txReadBuffer”。baseReadTx。缓冲器。创建“concurrentReadTx”时：
	// 如果缓存是最新的，则可以跳过“readTx.baseReadTx.buf”副本如果缓存是空的或过时的，则需要“readTx.baseReadTx.buf”副本
	txReadBufferCache txReadBufferCache

	stopc chan struct{}
	donec chan struct{}

	hooks Hooks

	lg *zap.Logger
}

type BackendConfig struct {
	// BoltDB数据库文件的路径
	Path string
	// 提交两次批量事务的最大时间差，用来初始化backend实例中的batchlnterval字段，默认值是1OOms。
	BatchInterval time.Duration
	// 指定每个批量读写事务能包含的最多的操作个数，当超过这个阈值之后，
	// 当前批量读写事务会自动提交。该字段用来初始化backend中的batchLimit字段，默认值是10000。
	BatchLimit int
	// BackendFreelistType是后端boltdb的自由列表类型。
	BackendFreelistType bolt.FreelistType
	// BoltDB使用了mmap技术对数据库文件进行映射，
	// 该字段用来设置mmap中使用的内存大小，该字段会在创建BoltDB实例时使用。
	MmapSize uint64
	// 记录器记录后端操作。
	Logger *zap.Logger
	// 取消同步将禁用所有fsync的使用。
	UnsafeNoFsync bool `json:"unsafe-no-fsync"`
	// Mlock阻止交换后端数据库文件
	Mlock bool

	// 在后端事务的生命周期中执行挂钩。
	Hooks Hooks
}

func DefaultBackendConfig() BackendConfig {
	return BackendConfig{
		BatchInterval: defaultBatchInterval,
		BatchLimit:    defaultBatchLimit,
		MmapSize:      initialMmapSize,
	}
}

func New(bcfg BackendConfig) Backend {
	return newBackend(bcfg)
}

// 初始化backend实例
// 使用了默认的配置参数井且其底层是调用newBackend()实现的，这些配置参数被封装成了BackendConfig结构体
func NewDefaultBackend(path string) Backend {
	bcfg := DefaultBackendConfig()
	bcfg.Path = path
	return newBackend(bcfg)
}

// 初始化backend实例
func newBackend(bcfg BackendConfig) *backend {
	if bcfg.Logger == nil {
		bcfg.Logger = zap.NewNop()
	}
	// 初始化BoltDB时的参数
	bopts := &bolt.Options{}
	if boltOpenOptions != nil {
		*bopts = *boltOpenOptions
	}
	bopts.InitialMmapSize = bcfg.mmapSize()	// mmap使用的内存大小
	bopts.FreelistType = bcfg.BackendFreelistType
	bopts.NoSync = bcfg.UnsafeNoFsync
	bopts.NoGrowSync = bcfg.UnsafeNoFsync
	bopts.Mlock = bcfg.Mlock

	db, err := bolt.Open(bcfg.Path, 0600, bopts)	// 创建boltDB实例
	if err != nil {
		bcfg.Logger.Panic("failed to open database", zap.String("path", bcfg.Path), zap.Error(err))
	}

	b := &backend{	// 创建backend实例，并初始化其中各个字段
		db: db,

		batchInterval: bcfg.BatchInterval,
		batchLimit:    bcfg.BatchLimit,
		mlock:         bcfg.Mlock,
		// 创建readTx实例并初始化backend.readTx字段
		readTx: &readTx{
			baseReadTx: baseReadTx{
				buf: txReadBuffer{
					txBuffer:   txBuffer{make(map[BucketID]*bucketBuffer)},
					bufVersion: 0,
				},
				buckets: make(map[BucketID]*bolt.Bucket),
				txWg:    new(sync.WaitGroup),
				txMu:    new(sync.RWMutex),
			},
		},
		txReadBufferCache: txReadBufferCache{
			mu:         sync.Mutex{},
			bufVersion: 0,
			buf:        nil,
		},

		stopc: make(chan struct{}),
		donec: make(chan struct{}),

		lg: bcfg.Logger,
	}
	// 创建batchTxBuffered实例并初始化backend.batchTx手段
	b.batchTx = newBatchTxBuffered(b)
	// 我们在newBatchTxBuffered之后设置它以跳过“空”提交。
	b.hooks = bcfg.Hooks
	// 启动一个单独的goroutine，其中会定时提交当前的批量读写事务，并开启新的批量读写事务
	go b.run()
	return b
}

// 返回了backend中的batchTx宇段
func (b *backend) BatchTx() BatchTx {
	return b.batchTx
}

// 返回了backend中的readTx宇段
func (b *backend) ReadTx() ReadTx { return b.readTx }

// ConcurrentReadTx创建并返回一个新的ReadTx，该ReadTx:
// a）创建并保留后端的副本。readTx。txReadBuffer，
// B）引用当前批处理间隔的boltdb read Tx（及其bucket缓存）。
func (b *backend) ConcurrentReadTx() ReadTx {
	b.readTx.RLock()
	defer b.readTx.RUnlock()
	// 在存储读取发送完成之前，防止boltdb read Tx被回滚。需要在按住readTx时调用。RLock（）。
	b.readTx.txWg.Add(1)

	// TODO:可能希望延迟复制读取缓冲区-在A）写入事务结束B）批处理间隔结束时创建副本。

	// 如果没有对缓存进行更新，则检查/更新缓存最近性
	// 如果没有缓存更新

	// 这一行，“ConcurrentReadTx”代码路径已经针对并发“写回”操作提供了保护
	// 这需要写锁来更新“readTx.baseReadTx.buf”。
	// 这意味着将“buf*txReadBuffer”设置为“readTx.buf.unsfeacopy（）”保证是最新的，而“txReadBufferCache.buf”可能因并发“写回”操作而过时。
	// 如果我们知道“buf*txReadBuffer”是最新的，我们只更新“txReadBufferCache.buf”。
	// 对“txReadBufferCache.buf”的更新将有利于以下“ConcurrentReadTx”创建
	// 避免复制“readTx.baseReadTx.buf”。
	b.txReadBufferCache.mu.Lock()

	curCache := b.txReadBufferCache.buf
	curCacheVer := b.txReadBufferCache.bufVersion
	curBufVer := b.readTx.buf.bufVersion

	isEmptyCache := curCache == nil
	isStaleCache := curCacheVer != curBufVer

	var buf *txReadBuffer
	switch {
	case isEmptyCache:
		// 在按住“b.txReadBufferCache.mu.Lock”的同时执行缓冲区的安全复制
		// 这只应该运行一次，因此不会有太多开销
		curBuf := b.readTx.buf.unsafeCopy()
		buf = &curBuf
	case isStaleCache:
		// 以最大化并发性，尝试不安全地复制缓冲区
		// 在复制缓冲区时释放锁--缓存可能再次变旧，
		// 被其他人覆盖。
		// 因此，我们需要再次检查readTx缓冲区版本
		b.txReadBufferCache.mu.Unlock()
		curBuf := b.readTx.buf.unsafeCopy()
		b.txReadBufferCache.mu.Lock()
		buf = &curBuf
	default:
		// 缓存既不空也不陈旧，只需使用当前缓冲区
		buf = curCache
	}
	// txReadBufferCache即可。当我们做一个不完美的（）
	// 时，可以修改bufVersion，因此，curCacheVer可能不再与
	// txReadBufferCache相同。B版本
	// 如果！isEmptyCache&&curCacheVer！=b、 txReadBufferCache。bufVersion
	// 然后，在复制“readTx.baseReadTx.buf”时，缓存已过时。
	// 不更新“txReadBufferCache.buf”是安全的，因为接下来的
	// “ConcurrentReadTx”创建将触发新的“readTx.baseReadTx.buf”副本
	// 并且“buf”仍然用于当前的“ConcurrentReadTx.baseReadTx.buf”。
	if isEmptyCache || curCacheVer == b.txReadBufferCache.bufVersion {
		// 如果从未设置缓存或没有人修改缓存，则继续
		b.txReadBufferCache.buf = buf
		b.txReadBufferCache.bufVersion = curBufVer
	}

	b.txReadBufferCache.mu.Unlock()

	// concurrentReadTx不应写入其txReadBuffer
	return &concurrentReadTx{
		baseReadTx: baseReadTx{
			buf:     *buf,
			txMu:    b.readTx.txMu,
			tx:      b.readTx.tx,
			buckets: b.readTx.buckets,
			txWg:    b.readTx.txWg,
		},
	}
}

// backend.ForceCommit（）方法会提交当前的读写事务并立即开启新的读写事务
func (b *backend) ForceCommit() {
	b.batchTx.Commit()
}

// backend.Snapshot（）方法的主要功能是用当前的BoltDB中的数据创建相应的快照，
// 其中使用前面提到的Tx.WriteTo（）方法备份整个BoltDB数据库的数据
func (b *backend) Snapshot() Snapshot {
	// 提交当前的读写事务，主要是为了提交缓冲区中的操作
	b.batchTx.Commit()

	b.mu.RLock()
	defer b.mu.RUnlock()
	tx, err := b.db.Begin(false)	// 开启一个只读事务
	if err != nil {
		b.lg.Fatal("failed to begin tx", zap.Error(err))
	}

	stopc, donec := make(chan struct{}), make(chan struct{})
	dbBytes := tx.Size()	// 获取整个BoltDB中保存的数据
	go func() {	// 启动一个单独的goroutine，用来检测快照数据是否已经发送完成
		defer close(donec)
		// sendRateBytes基于通过1 Gb/s连接传输快照数据
		// 假设最小tcp吞吐量为100MB/s。
		var sendRateBytes int64 = 100 * 1024 * 1024	// 这里假设发送快照的最小速度是1OOMB/s
		// 创建定时器
		warningTimeout := time.Duration(int64((float64(dbBytes) / float64(sendRateBytes)) * float64(time.Second)))
		if warningTimeout < minSnapshotWarningTimeout {
			warningTimeout = minSnapshotWarningTimeout
		}
		start := time.Now()
		ticker := time.NewTicker(warningTimeout)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:	// 超时未发送完快照数据，则会输出警告日志
				b.lg.Warn(
					"snapshotting taking too long to transfer",
					zap.Duration("taking", time.Since(start)),
					zap.Int64("bytes", dbBytes),
					zap.String("size", humanize.Bytes(uint64(dbBytes))),
				)

			case <-stopc:	// 发送快照数据结束
				snapshotTransferSec.Observe(time.Since(start).Seconds())
				return
			}
		}
	}()

	return &snapshot{tx, stopc, donec}	// 创建快照实例
}

func (b *backend) Hash(ignores func(bucketName, keyName []byte) bool) (uint32, error) {
	h := crc32.New(crc32.MakeTable(crc32.Castagnoli))

	b.mu.RLock()
	defer b.mu.RUnlock()
	err := b.db.View(func(tx *bolt.Tx) error {
		c := tx.Cursor()
		for next, _ := c.First(); next != nil; next, _ = c.Next() {
			b := tx.Bucket(next)
			if b == nil {
				return fmt.Errorf("cannot get hash of bucket %s", string(next))
			}
			h.Write(next)
			b.ForEach(func(k, v []byte) error {
				if ignores != nil && !ignores(next, k) {
					h.Write(k)
					h.Write(v)
				}
				return nil
			})
		}
		return nil
	})

	if err != nil {
		return 0, err
	}

	return h.Sum32(), nil
}

// backend.Size()方法返回的是backend.size字段值
func (b *backend) Size() int64 {
	return atomic.LoadInt64(&b.size)
}

func (b *backend) SizeInUse() int64 {
	return atomic.LoadInt64(&b.sizeInUse)
}

// backend.run（）方法中，会按照batchlnterval指定的时间间隔，
// 定时提交批量读写数据，在提交之后会立即开启一个新的批量读写事务。
func (b *backend) run() {
	defer close(b.donec)
	t := time.NewTimer(b.batchInterval)	// 创建定时器
	defer t.Stop()
	for {
		select {	// 阻塞等待上述定时器到期
		case <-t.C:
		case <-b.stopc:
			b.batchTx.CommitAndStop()	//
			return
		}
		if b.batchTx.safePending() != 0 {
			b.batchTx.Commit()	// 提交当前的批量读写事务，并开启一个新的批量读写事务
		}
		t.Reset(b.batchInterval)	// 重置定时器
	}
}

func (b *backend) Close() error {
	close(b.stopc)
	<-b.donec
	return b.db.Close()
}

// backend.Commits（）方法返回的是backend.commits字段
func (b *backend) Commits() int64 {
	return atomic.LoadInt64(&b.commits)
}

// backend.De企ag（）方法的主要功能就是整理当前BoltDB实例中的碎片，
// 其实就是提高其中Bucket的填充率。
// 整理碎片实际上是创建新的BoltDB数据库文件并将旧数据库文件的数据写入新数据库文件中。
// 因为在写入新数据库文件时是顺序写入的，所以会提高填充比例（FillPercent），
// 从而达到整理碎片的目的。需要注意的是，在整理碎片的过程中，需要持有readTx、batchTx和backend的锁
func (b *backend) Defrag() error {
	return b.defrag()	// 调用defrag（）方法完成具体的碎片整理过程
}

func (b *backend) defrag() error {
	now := time.Now()
	isDefragActive.Set(1)
	defer isDefragActive.Set(0)
	// 加锁，这里会获取readTx、batchTx和backend中的三把锁
	b.batchTx.Lock()
	defer b.batchTx.Unlock()
	b.mu.Lock()
	defer b.mu.Unlock()
	b.readTx.Lock()
	defer b.readTx.Unlock()
	// 提交当前的批量读写事务，注意参数，此次提交后不会立即打开新的批量读写事务
	b.batchTx.unsafeCommit(true)
	b.batchTx.tx = nil

	// 创建新的bolt.DB实例，对应的数据库文件是个临时文件
	dir := filepath.Dir(b.db.Path())
	temp, err := os.CreateTemp(dir, "db.tmp.*")
	if err != nil {
		return err
	}
	options := bolt.Options{}
	if boltOpenOptions != nil {
		options = *boltOpenOptions
	}
	options.OpenFile = func(_ string, _ int, _ os.FileMode) (file *os.File, err error) {
		return temp, nil
	}
	// 无论打开选项如何，都不要将tmp db加载到内存中
	options.Mlock = false
	tdbp := temp.Name()	// 获取新数据库文件的路径
	tmpdb, err := bolt.Open(tdbp, 0600, &options)
	if err != nil {
		return err
	}

	dbp := b.db.Path()	// 获取旧数据库文件的路径
	size1, sizeInUse1 := b.Size(), b.SizeInUse()
	if b.lg != nil {
		b.lg.Info(
			"defragmenting",
			zap.String("path", dbp),
			zap.Int64("current-db-size-bytes", size1),
			zap.String("current-db-size", humanize.Bytes(uint64(size1))),
			zap.Int64("current-db-size-in-use-bytes", sizeInUse1),
			zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse1))),
		)
	}
	// 进行碎片整理，其底层是创建一个新的BoltDB数据库文件并将当前数据库中的全部数据写入到其中，
	// 在写入过程中，会将新Bucket的填充比例设置成90%，从而达到碎片整理的效果
	err = defragdb(b.db, tmpdb, defragLimit)
	if err != nil {
		tmpdb.Close()
		if rmErr := os.RemoveAll(tmpdb.Path()); rmErr != nil {
			b.lg.Error("failed to remove db.tmp after defragmentation completed", zap.Error(rmErr))
		}
		return err
	}

	err = b.db.Close()	// 关闭旧的bolt.DB实例
	if err != nil {
		b.lg.Fatal("failed to close database", zap.Error(err))
	}
	err = tmpdb.Close()	// 关闭新的bolt.DB实例
	if err != nil {
		b.lg.Fatal("failed to close tmp database", zap.Error(err))
	}
	// 重命名新数据库文件，覆盖旧数据库文件
	err = os.Rename(tdbp, dbp)
	if err != nil {
		b.lg.Fatal("failed to rename tmp database", zap.Error(err))
	}

	defragmentedBoltOptions := bolt.Options{}
	if boltOpenOptions != nil {
		defragmentedBoltOptions = *boltOpenOptions
	}
	defragmentedBoltOptions.Mlock = b.mlock
	// 重新创建bolt.DB实例，此时使用的数据库文件是整理之后的新数据库文件
	b.db, err = bolt.Open(dbp, 0600, &defragmentedBoltOptions)
	if err != nil {
		b.lg.Fatal("failed to open database", zap.String("path", dbp), zap.Error(err))
	}
	// 开启新的批量读写事务以及只读事务
	b.batchTx.tx = b.unsafeBegin(true)

	b.readTx.reset()
	b.readTx.tx = b.unsafeBegin(false)

	size := b.readTx.tx.Size()
	db := b.readTx.tx.DB()
	atomic.StoreInt64(&b.size, size)
	atomic.StoreInt64(&b.sizeInUse, size-(int64(db.Stats().FreePageN)*int64(db.Info().PageSize)))

	took := time.Since(now)
	defragSec.Observe(took.Seconds())

	size2, sizeInUse2 := b.Size(), b.SizeInUse()
	if b.lg != nil {
		b.lg.Info(
			"finished defragmenting directory",
			zap.String("path", dbp),
			zap.Int64("current-db-size-bytes-diff", size2-size1),
			zap.Int64("current-db-size-bytes", size2),
			zap.String("current-db-size", humanize.Bytes(uint64(size2))),
			zap.Int64("current-db-size-in-use-bytes-diff", sizeInUse2-sizeInUse1),
			zap.Int64("current-db-size-in-use-bytes", sizeInUse2),
			zap.String("current-db-size-in-use", humanize.Bytes(uint64(sizeInUse2))),
			zap.Duration("took", took),
		)
	}
	return nil
}

// 重载方法主要完成了从旧数据库文件向新数据库文件的复制键值对的功能
func defragdb(odb, tmpdb *bolt.DB, limit int) error {
	// 在新数据库实例上开启一个读写事务，
	tmptx, err := tmpdb.Begin(true)
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tmptx.Rollback()
		}
	}()

	// 在旧数据库实例上开启一个只读事务，
	tx, err := odb.Begin(false)
	if err != nil {
		return err
	}
	defer tx.Rollback()	// 方法结束时关闭该只读事务
	// 获取旧实例上的Cursor，用于追历其中的所有Bucket
	c := tx.Cursor()

	count := 0
	for next, _ := c.First(); next != nil; next, _ = c.Next() {
		// 读取旧数据库实例中的所有Bucket，并在新数据库实例上创建对应Bucket
		b := tx.Bucket(next)	// 获取指定的Bucket实例，
		if b == nil {
			return fmt.Errorf("backend: cannot defrag bucket %s", string(next))
		}

		tmpb, berr := tmptx.CreateBucketIfNotExists(next)
		if berr != nil {
			return berr
		}
		// 为提高利用卒，将填充比例设立成90%，因为下面会从读取旧Bucket中全部的键位对，
		// 并填充到新Bucket中，这个过程是顺序写入的
		tmpb.FillPercent = 0.9 // 用于每个
		if err = b.ForEach(func(k, v []byte) error {	// 遍历旧Bucket中的全部键值对
			count++
			if count > limit {	// 当读取的键值对数量超过l凋伍，则提交当前读写事务（新数据库）
				err = tmptx.Commit()
				if err != nil {
					return err
				}
				// 重新开启一个读写事务，继续后面的写入操作，
				tmptx, err = tmpdb.Begin(true)
				if err != nil {
					return err
				}
				tmpb = tmptx.Bucket(next)
				tmpb.FillPercent = 0.9 // 设置填充比例

				count = 0
			}
			return tmpb.Put(k, v)	// 将读取到的键位对写入新数据库文件中
		}); err != nil {
			return err
		}
	}

	return tmptx.Commit()	// 最后提交读写事务（新数据库）
}

// 开启新的只读事务或读写事务都是通过backend.begin（）方法实现的，
// 该方法除了会开启事务（根据参数决定开启哪种类型的事务），还会更新backend中的相关字段，
func (b *backend) begin(write bool) *bolt.Tx {
	b.mu.RLock()
	tx := b.unsafeBegin(write)	// 开启事务
	b.mu.RUnlock()

	size := tx.Size()
	db := tx.DB()
	stats := db.Stats()
	// 更新backend.s工ze字段，记录了当前数据库的大小
	atomic.StoreInt64(&b.size, size)
	atomic.StoreInt64(&b.sizeInUse, size-(int64(stats.FreePageN)*int64(db.Info().PageSize)))
	atomic.StoreInt64(&b.openReadTxN, int64(stats.OpenTxN))

	return tx
}

func (b *backend) unsafeBegin(write bool) *bolt.Tx {
	tx, err := b.db.Begin(write)	// 调用了BoltDB的API开启事务
	if err != nil {
		b.lg.Fatal("failed to begin tx", zap.Error(err))
	}
	return tx
}

func (b *backend) OpenReadTxN() int64 {
	return atomic.LoadInt64(&b.openReadTxN)
}

type snapshot struct {
	*bolt.Tx
	stopc chan struct{}
	donec chan struct{}
}

func (s *snapshot) Close() error {
	close(s.stopc)
	<-s.donec
	return s.Tx.Rollback()
}
